package com.源码分析;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.junit.Test;

import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

public class ThreadDemo2 {

	private String getStringSync() {
	    try {
	    	System.out.println("callable: " + Thread.currentThread().getName());
	        TimeUnit.SECONDS.sleep(2);
	    } catch (InterruptedException e) {
	        e.printStackTrace();
	    }
	    return "Hello, Reactor!";
	}

	@Test
	public void testSyncToAsync() throws InterruptedException {
	    CountDownLatch countDownLatch = new CountDownLatch(1);
	    Mono.fromCallable(() -> getStringSync())    // 使用fromCallable声明一个基于Callable的Mono
	            .subscribeOn(Schedulers.elastic())  // 使用subscribeOn将任务调度到Schedulers内置的弹性线程池执行，弹性线程池会为Callable的执行任务分配一个单独的线程
	            .publishOn(Schedulers.newElastic("publishOn"))
	            .subscribe(x -> {
	            	System.out.println("subscribe: " + Thread.currentThread().getName());
	            	System.out.println(x);
	            }, null, () -> {
	            	System.out.println("complate: " + Thread.currentThread().getName());
	            	countDownLatch.countDown();
	            });
	    System.out.println(Thread.currentThread().getName());
	    countDownLatch.await();
	}

}
